package com.example.elevator.consumer;


import com.example.elevator.constants.KafkaConstants;
import com.example.elevator.models.Container;
import com.example.elevator.models.ElevatorStatus;
import com.example.elevator.models.Storage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;

@Component
public class ContainerListener {
    @Autowired
    private ExecutorService executer;

    @Autowired
    private KafkaTemplate<String, ElevatorStatus> kafkaTemplate;

    @Autowired
    private KafkaTemplate<String, Storage> kafkaTemplate2;

    @KafkaListener(
            topics = KafkaConstants.AGV_TOPIC,
            groupId = KafkaConstants.GROUP_ID
    )
    public void listen(Container container_msg) {
        System.out.println("sending via kafka listener..");

        executer.execute(new ElevatorTask(container_msg));
    }

    private class ElevatorTask implements Runnable {
        	Container entry_msg;

            public ElevatorTask(Container container) {
                entry_msg = container;
            }

            /*{
              "elevatorId": "Elevator-123",
              "containerId": "Container456",
              "status": "Moving",
              "from": "",
              "to": "Area B - Row 2 - Position 3",
              "timestamp": "2024-06-04T12:30:45Z"
            }
            */

            public void run() {
                int numThreads = 5;
                int randomSeconds = 5;
                try {
                    long threadId = Thread.currentThread().getId()%numThreads +1;
                    String elevatorId = "Elevator-"+threadId;
                    String containerId = entry_msg.getContainerNumber();
                    String status = "ON";
                    String from = "";
                    String to = entry_msg.getStorageLocation();

                    // 获取当前的日期和时间
                    LocalDateTime currentDateTime = LocalDateTime.now();

                    // 定义一个日期时间格式化器
                    DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

                    // 使用格式化器将日期时间转换为字符串
                    String timestamp = currentDateTime.format(formatter);

                    ElevatorStatus elevatorStatus = new ElevatorStatus(elevatorId, containerId, status, from, to, timestamp);

                    kafkaTemplate.send(KafkaConstants.ELEVATOR_TOPIC, elevatorStatus).get();

                    Random random = new Random();

                    // 生成一个0到6之间的随机数（包括0，不包括6）
                    randomSeconds = 15 + random.nextInt(25 - 15 + 1);
                    Thread.sleep(randomSeconds*1000);

                    // 模拟电梯到达目的地
                    status = "OFF";
                    timestamp = LocalDateTime.now().format(formatter);
                    elevatorStatus = new ElevatorStatus(elevatorId, containerId, status, to, "", timestamp);
                    kafkaTemplate.send(KafkaConstants.ELEVATOR_TOPIC, elevatorStatus).get();

                    //传递仓库存储信息
                    String containerNumber= entry_msg.getContainerNumber();
                    String storageLocation = entry_msg.getStorageLocation();
                    String goodsDescription = entry_msg.getGoodsDescription();
                    int maxCapacity = Storage.getMaxCapacity();
                    int currentUsage = 0;
                    if (entry_msg.getQuantity()<=maxCapacity){
                        currentUsage = entry_msg.getQuantity();
                    } else {//货物数量超出容量
                        goodsDescription = "Exceed maxCapacity!";
                        currentUsage = -1;//存储失败，通知仓库管理系统重新分配货物存储信息
                    }

                    Storage storage_msg = new Storage(containerNumber, storageLocation, goodsDescription, currentUsage);
                    kafkaTemplate2.send(KafkaConstants.ELEVATOR_TOPIC, storage_msg).get();

                    System.out.println("Storage information: " + storage_msg);



                } catch (InterruptedException | ExecutionException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }

            }
    }
}
